package com.gitee.wsl.struct.window

import com.gitee.wsl.ext.base.currentTimeMillis
import kotlinx.atomicfu.atomicArrayOfNulls
import kotlinx.atomicfu.locks.ReentrantLock

/**
 *
 *
 * Basic data structure for statistic metrics in Sentinel.
 *
 *
 *
 * Leap array use sliding window algorithm to count data. Each bucket cover `windowLengthInMs` time span,
 * and the total time span is [.intervalInMs], so the total bucket amount is:
 * `sampleCount = intervalInMs / windowLengthInMs`.
 *
 *
 * @param <T> type of statistic data
 * @author jialiang.linjl
 * @author Eric Zhao
 * @author Carpenter Lee
 *
 *
 * 特别声明: LeapArray的设计实现是使用了Sentinelv1.8.0版本的的LeapArray
 * see https://github.com/alibaba/Sentinel/blob/v1.8.0/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/statistic/base/LeapArray.java
</T> */

abstract class LeapArray<T>(
    /**
     * Get sample count (total amount of buckets).
     *
     * @return sample count
     */
    sampleCount: Int,
    /**
     * Get total interval length of the sliding window in milliseconds.
     *
     * @return interval in second
     */
    var intervalInMs: Int,
) {
    protected var windowLengthInMs: Int = intervalInMs / sampleCount

    /**
     * Get total interval length of the sliding window.
     *
     * @return interval in second
     */
    val intervalInSecond: Double = intervalInMs / 1000.0

    protected val array = atomicArrayOfNulls<WindowWrap<T>>(
        sampleCount
    )

    /**
     * The conditional (predicate) update lock is used only when current bucket is deprecated.
     */
    private val updateLock = ReentrantLock()

    /**
     * Get the bucket at current timestamp.
     *
     * @return the bucket at current timestamp
     */
    fun currentWindow(): WindowWrap<T>? {
        return currentWindow(currentTimeMillis)
    }

    /**
     * Create a new statistic value for bucket.
     *
     * @param timeMillis current time in milliseconds
     * @return the new empty bucket
     */
    abstract fun newEmptyBucket(timeMillis: Long): T

    /**
     * Reset given bucket to provided start time and reset the value.
     *
     * @param startTime  the start time of the bucket in milliseconds
     * @param windowWrap current bucket
     * @return new clean bucket at given start time
     */
    protected abstract fun resetWindowTo(
        windowWrap: WindowWrap<T>?,
        startTime: Long,
    ): WindowWrap<T>?

    fun calculateTimeIdx(
        /*@Valid*/
        timeMillis: Long,
    ): Int {
        val timeId = timeMillis / windowLengthInMs
        // Calculate current index so we can map the timestamp to the leap array.
        return (timeId % array.size).toInt()
    }

    protected fun calculateWindowStart(
        /*@Valid*/
        timeMillis: Long,
    ): Long {
        return timeMillis - timeMillis % windowLengthInMs
    }

    /**
     * Get bucket item at provided timestamp.
     *
     * @param timeMillis a valid timestamp in milliseconds
     * @return current bucket item at provided timestamp if the time is valid; null if time is invalid
     */
    fun currentWindow(timeMillis: Long): WindowWrap<T>? {
        if (timeMillis < 0) {
            return null
        }

        // 计算出窗口在哪个 元素位置处
        val idx = calculateTimeIdx(timeMillis)

        // 计算窗口的开始时间
        val windowStart = calculateWindowStart(timeMillis)

        /*
         * Get bucket item at given time from the array.
         *
         * (1) Bucket is absent, then just create a new bucket and CAS update to circular array.
         * (2) Bucket is up-to-date, then just return the bucket.
         * (3) Bucket is deprecated, then reset current bucket and clean all deprecated buckets.
         */
        while (true) {
            val old = array[idx].value
            if (old == null) {
                /*
                 *     B0       B1      B2    NULL      B4
                 * ||_______|_______|_______|_______|_______||___
                 * 200     400     600     800     1000    1200  timestamp
                 *                             ^
                 *                          time=888
                 *            bucket is empty, so create new and update
                 *
                 * If the old bucket is absent, then we create a new bucket at {@code windowStart},
                 * then try to update circular array via a CAS operation. Only one thread can
                 * succeed to update, while other threads yield its time slice.
                 */
                val window = WindowWrap(windowLengthInMs.toLong(), windowStart, newEmptyBucket(timeMillis))

                if (array[idx].compareAndSet(null,window)) {
                    // Successfully updated, return the created bucket.
                    return window
                } else {
                    // Contention failed, the thread will yield its time slice to wait for bucket available.
                    Thread.yield()
                }
            } else if (windowStart == old.windowStart) {
                /*
                 *     B0       B1      B2     B3      B4
                 * ||_______|_______|_______|_______|_______||___
                 * 200     400     600     800     1000    1200  timestamp
                 *                             ^
                 *                          time=888
                 *            startTime of Bucket 3: 800, so it's up-to-date
                 *
                 * If current {@code windowStart} is equal to the start timestamp of old bucket,
                 * that means the time is within the bucket, so directly return the bucket.
                 */
                return old
            } else if (windowStart > old.windowStart) {
                /*
                 *   (old)
                 *             B0       B1      B2    NULL      B4
                 * |_______||_______|_______|_______|_______|_______||___
                 * ...    1200     1400    1600    1800    2000    2200  timestamp
                 *                              ^
                 *                           time=1676
                 *          startTime of Bucket 2: 400, deprecated, should be reset
                 *
                 * If the start timestamp of old bucket is behind provided time, that means
                 * the bucket is deprecated. We have to reset the bucket to current {@code windowStart}.
                 * Note that the reset and clean-up operations are hard to be atomic,
                 * so we need a update lock to guarantee the correctness of bucket update.
                 *
                 * The update lock is conditional (tiny scope) and will take effect only when
                 * bucket is deprecated, so in most cases it won't lead to performance loss.
                 */
                if (updateLock.tryLock()) {
                    try {
                        // Successfully get the update lock, now we reset the bucket.
                        return resetWindowTo(old, windowStart)
                    } finally {
                        updateLock.unlock()
                    }
                } else {
                    // Contention failed, the thread will yield its time slice to wait for bucket available.
                    Thread.yield()
                }
            } else if (windowStart < old.windowStart) {
                // Should not go through here, as the provided time is already behind.
                return WindowWrap(
                    windowLengthInMs.toLong(),
                    windowStart,
                    newEmptyBucket(timeMillis)
                )
            }
        }
    }

    /**
     * Get the previous bucket item before provided timestamp.
     *
     * @param timeMillis a valid timestamp in milliseconds
     * @return the previous bucket item before provided timestamp
     */
    fun getPreviousWindow(timeMillis: Long): WindowWrap<T>? {
        var timeMillis = timeMillis
        if (timeMillis < 0) {
            return null
        }
        val idx = calculateTimeIdx(timeMillis - windowLengthInMs)
        timeMillis -= windowLengthInMs
        val wrap = array[idx].value

        if (wrap == null || isWindowDeprecated(wrap)) {
            return null
        }

        if (wrap.windowStart + windowLengthInMs < (timeMillis)) {
            return null
        }

        return wrap
    }

    val previousWindow: WindowWrap<T>?
        /**
         * Get the previous bucket item for current timestamp.
         *
         * @return the previous bucket item for current timestamp
         */
        get() = getPreviousWindow(System.currentTimeMillis())

    /**
     * Get statistic value from bucket for provided timestamp.
     *
     * @param timeMillis a valid timestamp in milliseconds
     * @return the statistic value if bucket for provided timestamp is up-to-date; otherwise null
     */
    fun getWindowValue(timeMillis: Long): T? {
        if (timeMillis < 0) {
            return null
        }
        val idx = calculateTimeIdx(timeMillis)

        val bucket = array[idx].value

        if (bucket == null || !bucket.isTimeInWindow(timeMillis)) {
            return null
        }

        return bucket.value
    }

    /**
     * Check if a bucket is deprecated, which means that the bucket
     * has been behind for at least an entire window time span.
     *
     * @param windowWrap a non-null bucket
     * @return true if the bucket is deprecated; otherwise false
     */
    fun isWindowDeprecated(
        /*@NonNull*/
        windowWrap: WindowWrap<T>,
    ): Boolean {
        return isWindowDeprecated(System.currentTimeMillis(), windowWrap)
    }

    fun isWindowDeprecated(time: Long, windowWrap: WindowWrap<T>): Boolean {
        return time - windowWrap.windowStart > intervalInMs
    }

    /**
     * Get valid bucket list for entire sliding window.
     * The list will only contain "valid" buckets.
     *
     * @return valid bucket list for entire sliding window.
     */
    fun list(): List<WindowWrap<T>> {
        return list(System.currentTimeMillis())
    }

    fun list(validTime: Long): List<WindowWrap<T>> {
        val size = array.size
        val result: MutableList<WindowWrap<T>> = ArrayList(size)

        for (i in 0..<size) {
            val windowWrap = array[i].value
            if (windowWrap == null || isWindowDeprecated(validTime, windowWrap)) {
                continue
            }
            result.add(windowWrap)
        }

        return result
    }

    /**
     * Get all buckets for entire sliding window including deprecated buckets.
     *
     * @return all buckets for entire sliding window
     */
    fun listAll(): List<WindowWrap<T>> {
        val size = array.size
        val result: MutableList<WindowWrap<T>> = ArrayList(size)

        for (i in 0..<size) {
            val windowWrap = array[i].value ?: continue
            result.add(windowWrap)
        }

        return result
    }

    /**
     * Get aggregated value list for entire sliding window.
     * The list will only contain value from "valid" buckets.
     *
     * @return aggregated value list for entire sliding window
     */
    fun values(): List<T> {
        return values(System.currentTimeMillis())
    }

    fun values(timeMillis: Long): List<T> {
        if (timeMillis < 0) {
            return ArrayList()
        }
        val size = array.size
        val result: MutableList<T> = ArrayList(size)

        for (i in 0..<size) {
            val windowWrap = array[i].value
            if (windowWrap == null || isWindowDeprecated(timeMillis, windowWrap)) {
                continue
            }
            result.add(windowWrap.value)
        }
        return result
    }

    /**
     * Get the valid "head" bucket of the sliding window for provided timestamp.
     * Package-private for test.
     *
     * @param timeMillis a valid timestamp in milliseconds
     * @return the "head" bucket if it exists and is valid; otherwise null
     */
    fun getValidHead(timeMillis: Long): WindowWrap<T>? {
        // Calculate index for expected head time.
        val idx = calculateTimeIdx(timeMillis + windowLengthInMs)

        val wrap = array[idx].value
        if (wrap == null || isWindowDeprecated(wrap)) {
            return null
        }

        return wrap
    }

    val validHead: WindowWrap<T>?
        /**
         * Get the valid "head" bucket of the sliding window at current timestamp.
         *
         * @return the "head" bucket if it exists and is valid; otherwise null
         */
        get() = getValidHead(System.currentTimeMillis())

    fun debug(time: Long) {
        val sb = StringBuilder()
        val lists = list(time)
        sb.append("Thread_").append(Thread.currentThread().id).append("_")
        for (window in lists) {
            sb.append(window.windowStart).append(":").append(window.value.toString())
        }
        println(sb.toString())
    }

    fun currentWaiting(): Long {
        // TODO: default method. Should remove this later.
        return 0
    }

    fun addWaiting(time: Long, acquireCount: Int) {
        // Do nothing by default.
        throw UnsupportedOperationException()
    }
}
